package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSqlApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(
                10016, 4, Constant.TOPIC_DWD_TRADE_ORDER_PAY_SUC_DETAIL, Constant.TOPIC_DWD_TRADE_ORDER_PAY_SUC_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //获取下单事务事实表
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "order_status string,\n" +
                "out_trade_no string,\n" +
                "trade_body string,\n" +
                "session_id string,\n" +
                "province_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "origin_amount string,\n" +
                "coupon_amount string,\n" +
                "final_amount string,\n" +
                "create_time string,\n" +
                "update_time  string ,\n" +
                "ts bigint\n" +
                ")" + SqlUtil.getKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, Constant.TOPIC_DWD_TRADE_ORDER_PAY_SUC_DETAIL)
        );

        //tableEnv.sqlQuery("select * from dwd_trade_order_detail ").execute().print();

        //读取topic_db数据
        getTopicDbTable(tableEnv, Constant.TOPIC_DWD_TRADE_ORDER_PAY_SUC_DETAIL);
        //获取字典表
        //getBaseTable(tableEnv);
        //过滤支付表数据
        Table paymentInfo = tableEnv.sqlQuery("select \n" +
                "`table`,\n" +
                "data['id']                 id,\n" +
                "data['out_trade_no']       out_trade_no,\n" +
                "data['order_id']           order_id,\n" +
                "data['alipay_trade_no']    alipay_trade_no,\n" +
                "data['total_amount']       total_amount,\n" +
                "data['trade_body']         trade_body,\n" +
                "data['payment_type']       payment_type,\n" +
                "data['payment_status']     payment_status,\n" +
                "data['create_time']        create_time,\n" +
                "data['update_time']        update_time,\n" +
                "data['callback_content']   callback_content,\n" +
                "data['callback_time']      callback_time,\n" +
                "ts \n" +
                "from " +
                Constant.TOPIC_DB + "\n" +
                "where  `table`='payment_info' \n" +
                "and `type`='insert' " +
                "and data['payment_status'] = '1602'");
        //paymentInfo.execute().print();
        tableEnv.createTemporaryView("payment_info", paymentInfo);
        //将
        Table table  = tableEnv.sqlQuery("select \n" +
                "dod.id                 `id`,\n" +
                "dod.user_id            user_id,\n" +
                "dod.order_status       order_status,\n" +
                "dod.out_trade_no       out_trade_no,\n" +
                "dod.trade_body         trade_body,\n" +
                "dod.session_id         session_id,\n" +
                "dod.province_id        province_id,\n" +
                "dod.course_id          course_id,\n" +
                "dod.course_name        course_name,\n" +
                "dod.origin_amount      origin_amount,\n" +
                "dod.coupon_amount      coupon_amount,\n" +
                "dod.final_amount       final_amount,\n" +
                "dod.create_time        create_time,\n" +
                "dod.update_time        update_time,\n" +
                "pi.ts                  ts,\n" +
                "pi.payment_type        payment_type,\n" +
                "pi.callback_content    callback_content,\n" +
                "pi.callback_time       callback_time\n " +
                "from dwd_trade_order_detail dod\n" +
                "join payment_info pi \n" +
                "on dod.id = pi.order_id\n" );

        //table.execute().print();

        tableEnv.executeSql(
                "create table dwd_trade_order_pay_suc_detail (\n" +
                        "`id` string,\n" +
                        "`user_id` string ,\n" +
                        "`order_status` string,\n" +
                        "`out_trade_no` string,\n" +
                        "`trade_body` string,\n" +
                        "`session_id` string,\n" +
                        "`province_id` string,\n" +
                        "`course_id` string,\n" +
                        "`course_name` string,\n" +
                        "`origin_amount` string,\n" +
                        "`coupon_amount` string,\n" +
                        "`final_amount` string,\n" +
                        "`create_time` string,\n" +
                        "`update_time` string,\n" +
                        "`ts` bigint,\n" +
                        "`payment_type` string,\n" +
                        "`callback_content` string,\n" +
                        "`callback_time` string,\n" +
                        "primary key(id) not enforced \n" +
                        ")" + " WITH (\n" +
                        "  'connector' = 'upsert-kafka',\n" +
                        "  'topic' = '" + Constant.TOPIC_DWD_TRADE_ORDER_PAY_SUC_DETAIL + "',\n" +
                        "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                        "  'key.format' = 'json',\n" +
                        "  'value.format' = 'json'\n" +
                        ")");

        table.executeInsert("dwd_trade_order_pay_suc_detail");
    }
}
